package com.geekbg.zookeeper.distributedqueue;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.recipes.queue.*;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * @author pizhihui
 * @date 2023-02-28
 */
public class TestDistributedQueue {

    private static final String QUEUE_PATH = "/a/queue";

    private static final QueueSerializer<TestQueueItem> serializer = new QueueItemSerializer();

    @Test
    public void testPutListener() throws Exception {
        final int itemQty = 10;

        String connStr = "localhost:2181";

        DistributedQueue<TestQueueItem> queue = null;
        CuratorFramework client = CuratorFrameworkFactory.newClient(connStr, new RetryOneTime(1));
        client.start();

//        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL).inBackground(new BackgroundCallback() {
//            @Override
//            public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
//                System.out.println("结果..." + curatorEvent.getResultCode() + "..." + curatorEvent);
//            }
//        }).forPath("/b/queue/queu-", "123".getBytes());
//        System.in.read();

        final int RETRY_COUNT = 1;
        final List<String> names = new ArrayList<String>();
        final CountDownLatch retryCounter = new CountDownLatch(RETRY_COUNT + 1);


        try {

            QueueConsumer<TestQueueItem> consumer = new QueueConsumer<TestQueueItem>()
            {
                @Override
                public void consumeMessage(TestQueueItem messsage) throws Exception
                {
                    List<String> queueItems = client.getChildren().forPath(QUEUE_PATH);
                    System.out.println("开始消费数据了..." + queueItems.size());
                    names.add(queueItems.get(0));

                    if (retryCounter.getCount() > 5)
                    {
                        retryCounter.countDown();
                        throw new Exception("Something went wrong");
                    }
                    else
                    {
                        retryCounter.countDown();
                    }
                }

                @Override
                public void stateChanged(CuratorFramework client, ConnectionState newState)
                {
                }
            };
            BlockingQueueConsumer<TestQueueItem> blockingQueueConsumer = new BlockingQueueConsumer<TestQueueItem>(new ConnectionStateListener() {
                @Override
                public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                    System.out.println("改变. ..........");
                }
            });

            queue = QueueBuilder.builder(client, consumer, serializer, QUEUE_PATH).buildQueue();
            queue.start();

            QueueTestProducer producer = new QueueTestProducer(queue, itemQty, 0);

            final AtomicInteger listenerCalls = new AtomicInteger(0);
            QueuePutListener<TestQueueItem> listener = new QueuePutListener<TestQueueItem>() {
                @Override
                public void putCompleted(TestQueueItem item) {
                    listenerCalls.incrementAndGet();
                }

                @Override
                public void putMultiCompleted(MultiItem<TestQueueItem> items) {
                }
            };
            queue.getPutListenerContainer().addListener(listener);

            ExecutorService service = Executors.newCachedThreadPool();
            service.submit(producer);


            retryCounter.await(10, TimeUnit.SECONDS);
//            int iteration = 0;
//            while (consumer.size() < itemQty) {
//                if (++iteration < 10) {
//                    continue;
//                }
//                Thread.sleep(1000);
//            }
//
//            int i = 0;
//            for (TestQueueItem item : consumer.getItems()) {
//                System.out.println("item: " + item.str);
//                // assertEquals(item.str, Integer.toString(i++));
//            }

            // assertEquals(listenerCalls.get(), itemQty);
        } finally {
            CloseableUtils.closeQuietly(queue);
            CloseableUtils.closeQuietly(client);
        }
    }

}
